package com.hzgj.soa.spring;

import com.google.common.collect.Maps;
import com.hzgj.bcl.soa.ChainStack;
import com.hzgj.bcl.soa.api.IdempotentProcessor;
import com.hzgj.bcl.soa.config.ServerDef;
import com.hzgj.bcl.soa.config.ServiceDef;
import com.hzgj.bcl.soa.interceptor.IdempotentInterceptor;
import com.hzgj.bcl.soa.interceptor.SerializationInterceptor;
import com.hzgj.bcl.soa.registry.RegistryContext;
import com.hzgj.bcl.soa.rpc.IpAddress;
import com.hzgj.bcl.soa.rpc.server.RpcServer;
import com.hzgj.bcl.soa.rpc.server.ServerConfig;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.GlobalInvoke;
import com.hzgj.bcl.util.jmx.Management;
import com.hzgj.bcl.util.lang.StrKit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class ServerBean implements ApplicationContextAware {

    private Logger logger = LoggerFactory.getLogger(getClass());
    private ApplicationContext applicationContext;

    public void start() {
        Map<String, ServerDef> servers = applicationContext.getBeansOfType(ServerDef.class);
        if (servers.size() == 0) {
            logger.error("no server definition can be found.");
            return;
        }
        List<ServerDef> serverDefs = servers.values().stream().filter(serverDef -> serverDef.getServiceDefs().size() > 0).sorted((o1, o2) -> {
            if (o1.isSuspend() && !o2.isSuspend()) {
                return 1;
            } else if (!o1.isSuspend() && o2.isSuspend()) {
                return -1;
            }
            return 0;
        }).collect(Collectors.toList());
        if (serverDefs.size() == 0) {
            throw new RuntimeException("no service need to register.");
        }
        Map<IpAddress, Map<String, ServiceDef.Attr>> serverServiceMapping = Maps.newHashMap();
        serverDefs.stream().forEach(serverDef -> {
            Map<String, ServiceDef.Attr> serviceMapping = Maps.newHashMap();
            try {
                serverServiceMapping.put(new IpAddress(serverDef.getHost(), serverDef.getPort()), serviceMapping);
                GlobalInvoke.addContext(new GlobalInvoke.ServerStartedInvokeEvent(serverDef.getId()), this, ServerBean.class.getMethod("registryServer", RegistryContext.ServerRegisterModel.class),
                        new Object[]{RegistryContext.ServerRegisterModel.builder().serverId(serverDef.getId()).services(serviceMapping).build()});
            } catch (Exception e) {
                logger.error("{}", e);
            }
            buildServer(serverDef, serviceMapping);
        });
    }

    private void buildServer(ServerDef serverDef, Map<String, ServiceDef.Attr> serviceMapping) {
        ServerConfig serverConfig = new ServerConfig();
        serverConfig.setPort(serverDef.getPort());
        serverConfig.setHost(serverDef.getHost());
        serverConfig.setSuspend(serverDef.isSuspend());
        serverConfig.setName(serverDef.getId());
        serverConfig.setProtocol(serverDef.getProtocol());
        if (serverDef.getBossThreads() > 0) {
            serverConfig.setBossGroupThreads(serverDef.getBossThreads());
        }
        if (serverDef.getWorkerThreads() > 0) {
            serverConfig.setWorkerGroupThreads(serverDef.getWorkerThreads());
        }

        if (StrKit.notBlank(serverDef.getBizPool())) {
            ServerConfig.BizThreadPoolConfig bizThreadPoolConfig = applicationContext.getBean(serverDef.getBizPool(), ServerConfig.BizThreadPoolConfig.class);
            if (bizThreadPoolConfig.getMax() < bizThreadPoolConfig.getMin()) {
                throw new RuntimeException("pool[" + serverDef.getBizPool() + "max = " + bizThreadPoolConfig.getMax() + ",min = " + bizThreadPoolConfig.getMin
                        () + "],min need lower max.");
            }
            serverConfig.setBizThreadPoolConfig(bizThreadPoolConfig);
        }
        ChainStack chainStack = new ChainStack();
        chainStack.addInterceptor(new SerializationInterceptor());
        RpcServer rpcServer = new RpcServer(chainStack, serverConfig);
        Management.register(rpcServer, Constants.getRpcServerObjectName(Integer.toString(serverDef.getPort())));
        Map<String, Map<String, IdempotentProcessor>> serviceMethodIdempotents = Maps.newHashMap();
        serverDef.getServiceDefs().stream().forEach(attr -> {
            Object bean = null;
            if (StrKit.isBlank(attr.getRef())) {
                try {
                    bean = applicationContext.getBean(Class.forName(attr.getInterfaceClass()));
                } catch (ClassNotFoundException e) {

                }
            } else {
                bean = applicationContext.getBean(attr.getRef());
            }
            if (bean == null) {
                throw new RuntimeException("can't find object of " + attr.getInterfaceClass());
            }
            rpcServer.registerProcessor(attr.getInterfaceClass(), bean);
            if (attr.getMethodDefs() != null) {
                attr.getMethodDefs().values().forEach(methodDef -> {
                    if (methodDef.getIdempotentProcessorClass() != null) {
                        serviceMethodIdempotents.compute(attr.getInterface(), (key, idempotentProcessorMap) -> {
                            if (idempotentProcessorMap == null) {
                                idempotentProcessorMap = Maps.newHashMap();
                            }
                            idempotentProcessorMap.putIfAbsent(methodDef.getName(), applicationContext.getBean(methodDef.getIdempotentProcessorClass()));
                            return idempotentProcessorMap;
                        });
                    }
                });
            }
            serviceMapping.put(attr.getInterface(), attr);
        });
        if (serviceMethodIdempotents.size() > 0) {
            chainStack.addInterceptor(new IdempotentInterceptor(serviceMethodIdempotents));
        }
        try {
            rpcServer.start();
        } catch (Exception e) {
            logger.error("{}", e);
            try {
                rpcServer.close();
            } catch (IOException e1) {
            }
        }
    }

    public void registryServer(RegistryContext.ServerRegisterModel serverRegisterModel) {
        try {
            RegistryBean registryBean = applicationContext.getBean(RegistryBean.class);
            if (registryBean != null) {
                registryBean.registryServer(serverRegisterModel);
            }
        } catch (NoSuchBeanDefinitionException e1) {
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        com.hzgj.soa.spring.ServerConfig serverConfig = new com.hzgj.soa.spring.ServerConfig();
        serverConfig.afterPropertiesSet(applicationContext);
    }

}